#include <stdarg.h>
#include <unistd.h>
#include <arpa/inet.h>
#include "inet.h"
#include "event.h"
#include "thread_pool.h"
#include "block_queue.h"


#define MAX_DISPATHER_THREADS 8

typedef struct entity_t {
    queue_t queue;
    int fd; 
} entity_t;

typedef struct thread_data {
   pthread_t pid;
   int ep;
   int idx;
   int notify_fd;
   int receive_fd;
   block_queue_t free_conn;
   block_queue_t new_conn;
} thread_data_t;



int listen_fd;
//block_queue_t worker_entities;
//block_queue_t free_entities;
//pthread_t **threads;
thread_data_t **threads;
static int last_thread;


entity_t *  create_new_entity(){
    entity_t *en = (entity_t*)thread_calloc(sizeof(entity_t));
    if(!en){
      err_sys("create_new_entity thread_calloc error");
      return NULL;
    }
    printf("create_new_entity():%p:%p\n",&en->queue,en);
    return en;
}

entity_t * get_free_entity(block_queue_t *free_entities){
    entity_t *en = NULL;
    queue_t *qq = block_queue_pop(free_entities);
    if(qq)
      en = queue_data(qq,entity_t,queue);
    else
      en = create_new_entity();
    if(!en){
      //destoryAll();
      err_sys("get free entity error!\n");
    }
    return en;
}

entity_t * makeup_entity(block_queue_t *free_entities,int fd){
    entity_t *en = get_free_entity(free_entities);
    en->fd = fd; 
    return en;
}


void dispatch_conn_new(int new_fd){
   int tid = (last_thread + 1) % MAX_DISPATHER_THREADS;
   thread_data_t *data = threads[tid];
   last_thread = tid;

    
   entity_t *en = makeup_entity(&data->free_conn,new_fd);
   block_queue_push(&data->new_conn,&en->queue);

   if(write(data->notify_fd,"",1) != 1){
      err_ret("write thread notify pip");
   }
}




void * thread_worker(void *_idx)
{
  printf("thread_worker():create %ld worker success! and the global block_queue idx:%d\n",pthread_self(),*((int*)_idx));
  int ret  = 0;
  if((ret = thread_setspecific(_idx)) != 0){
    err_ret("%ld thread_setspecific() error",pthread_self());
    return (void*)0;
  }

  thread_data_t *data = threads[*((int*)_idx)];

  char buff[1];
  int  fdcnt , i;
  struct epoll_event events[MAX_DISPATHER_THREADS]; 
  for(;;){
    fdcnt = epoll_wait(data->ep,events,MAX_DISPATHER_THREADS,-1);
 
    for(i=0;i<fdcnt;i++){
       int fd = events[i].data.fd; 
       if(fd < 0){
         err_ret("worker invalid fd");
         continue;
       }
       
       if(fd == data->receive_fd){
          if(read(data->receive_fd,buff,1) != 1){
             err_ret("cant read from notify pipe"); 
          }else{
             queue_t *qq  = block_queue_pop(&data->new_conn);
             if(qq == NULL && data->new_conn.block_flag == 0) break; // thread exit;
             entity_t *en = queue_data(qq,entity_t,queue);
             printf("get new fd:%d\n",en->fd);
             epoll_add(data->ep,en->fd,EPOLLIN,EPOLLET); 
             block_queue_push(&data->free_conn,&en->queue);
          }
          continue;
       }
       //int ret = 0;
       /*if((ret=do_download(en)) != 0)
         download_notify(en,NULL,ret);*/
       int  len = 10;
       char *buff=NULL;
       printf("BUFF:%p\n",buff);
       int ret = epoll_read(fd,&buff,&len);
       if(ret > 0){
         printf("BUFF2:%p\n",buff);
         printf("receive:%s\nlen:%d\n",buff,len);
       }
       //thread_reset_pool();
       //reset_entity(en);
       //block_queue_push(&free_entities,&en->queue);
    }
  }
  return _idx;
}

void init_threads(int threadcnt)
{
  int i,ret;
  pthread_attr_t attr;
  ret = thread_key_create(free);
  if(ret != 0){
    //destoryAll();
    err_sys("thread_key_create() error");
  }
  ret = pthread_attr_init(&attr);
  ret = pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED);

  threads = (thread_data_t**)thread_calloc(sizeof(thread_data_t*)*threadcnt);
  for(i=0;i<threadcnt;i++){
    int fd[2];
    if(pipe(fd)){
       err_sys("cant create notify pipe");
    }

    thread_data_t *data = (thread_data_t*)thread_calloc(sizeof(thread_data_t));
    data->idx = i; 
    data->ep  = epoll_init(1024);
    data->notify_fd =  fd[1];
    data->receive_fd = fd[0];  
    block_queue_init(&data->free_conn,0,0);
    block_queue_init(&data->new_conn,0,0);

    epoll_add(data->ep,data->receive_fd,EPOLLIN,EPOLLET);

    threads[i] = data;
    int *idx = (int*)malloc(sizeof(int));
    *idx = i;
    ret = pthread_create(&data->pid,NULL,thread_worker,(void*)idx);
    if(ret != 0){
	//destoryAll();
        err_sys("pthread_create():cant create worker");
    }
  }
  pthread_attr_destroy(&attr);
}



int main(int argc ,char *argv[])
{
   int port;
   struct in_addr host = {INADDR_NONE};
  
   int ret = -1 , i , fdcnt; 
   while((ret=getopt(argc,argv,":h:p:")) != -1){
      switch(ret){
        case 'h':
          if(inet_pton(AF_INET,optarg,(void*)&host) != 1)
             err_sys("inet_pton error!");
          break;
        case 'p': 
          port = atoi(optarg);
          break;
        case '?':
          err_sys("invalid option");
          break;
        case ':':
          err_sys("no opt value");
          break;
      }
   }

   if(argc < 2)
      err_quit("Usage: msg  -p port [-h host]");

   if(host.s_addr == INADDR_NONE)
      host.s_addr = INADDR_ANY; 

   int ep =epoll_init(1024);
   
   listen_fd = open_socket(ntohl(host.s_addr),port);
   
   setnonblocking(listen_fd);
   epoll_add(ep,listen_fd,EPOLLIN,EPOLLET);

   printf("host:%d port:%d listen_fd:%d\n",host.s_addr,port,listen_fd);

   thread_init_pools(MAX_DISPATHER_THREADS,1024); 

   /*block_queue_init(&free_entities,0,0);
   for(i=0;i<MAX_DISPATHER_THREADS;i++){
      entity_t *en = create_new_entity(); 
      block_queue_push(&free_entities,&en->queue);
   }
   block_queue_init(&worker_entities,MAX_DISPATHER_THREADS,1);
   */


   init_threads(MAX_DISPATHER_THREADS);

   struct epoll_event events[MAX_DISPATHER_THREADS]; 
   for(;;){
      fdcnt = epoll_wait(ep,events,MAX_DISPATHER_THREADS,-1);
      for(i=0;i<fdcnt;i++){
         int fd = events[i].data.fd;
         if(fd < 0){
            err_ret("get invalidate fd");
            continue;
         }
         if(fd == listen_fd) {
           struct sockaddr_in client_addr;
           socklen_t  socklen;
           int clientfd = accept(listen_fd,(struct sockaddr*)&client_addr,&socklen);
           if(clientfd < 0){
              err_ret("accept new connect err:");
              continue;
           } 

           setnonblocking(clientfd);
         
           dispatch_conn_new(clientfd);
         }else{
            printf("unkown fd:%d\n",events[i].data.fd);
         }/*else if(events[i].events & EPOLLOUT)*/
      }
   }
  
   return 0;
}
